package io.xxx.omni.oss

import org.apache.pulsar.client.admin.PulsarAdmin
import org.apache.pulsar.client.api.*
import java.util.*
import java.util.concurrent.TimeUnit

class PulsarTests {
}

fun main() {
    val admin = PulsarAdmin.builder()
        .serviceHttpUrl("https://10.48.137.117:6650,10.48.137.118:6650,10.48.137.119:6650")
        .build()
//    val topic = "persistent://public/default/test"
    val topic = "persistent://public/default/document-aikucun"
//    val topics = admin.topics().getList("public/default")
//    topics.forEach {
//        println(it)
//    }
//    admin.topics().createPartitionedTopic(topic, 12)
//
//
    val client = PulsarClient.builder()
        .serviceUrl("pulsar://10.48.137.117:6650,10.48.137.118:6650,10.48.137.119:6650")
        .build()
    val producer = client.newProducer(Schema.STRING)
        .batchingMaxPublishDelay(200, TimeUnit.MILLISECONDS)
        .batchingMaxBytes(1024 * 1024 * 16)
        .topic(topic)
        .blockIfQueueFull(true)
        .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
        .compressionType(CompressionType.LZ4)
//        .batcherBuilder(BatcherBuilder.KEY_BASED)
        .create()

    for (i in 1..100) {
        producer.sendAsync("你好 - $i  ${UUID.randomUUID().toString().replace("-", "").repeat(500)}")
            .whenComplete { _, u ->
                u?.printStackTrace()
            }
    }
    TimeUnit.SECONDS.sleep(10)
}